package com.itmentu.listener;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @Author：http://www.itmentu.com
 * @date： 2021/12/18 17:04
 * @Description：
 ***/
@Component
public class MessageListener {

    @KafkaListener(topics = {"itmentu"},groupId = "itmentuGroup")
    public void listener(ConsumerRecord<String,String> record, Acknowledgment ack){
        //获取消息
        String message = record.value();
        //消息偏移量
        long offset = record.offset();
        System.out.println("读取的消息："+message+"\n当前偏移量："+offset);

        //手动提交偏移量
        ack.acknowledge();
    }


    /**
     * 消费者组mentuQueueGroup,方法listenerMentuQueue1
     * @param record
     */
    @KafkaListener(topics = {"mentuQueue"},groupId = "mentuQueueGroup")
    public void listenerMentuQueue1(ConsumerRecord<String,String> record){
        //获取消息
        String message = record.value();
        //消息偏移量
        long offset = record.offset();
        System.out.println("listenerMentuQueue1-->读取的消息："+message+"，分区："+record.partition()+"-->当前偏移量："+offset);
    }

    /**
     * 消费者组mentuQueueGroup,方法listenerMentuQueue2
     * @param record
     */
    @KafkaListener(topics = {"mentuQueue"},groupId = "mentuQueueGroup")
    public void listenerMentuQueue2(ConsumerRecord<String,String> record){
        //获取消息
        String message = record.value();
        //消息偏移量
        long offset = record.offset();
        System.out.println("listenerMentuQueue2-->读取的消息："+message+"，分区："+record.partition()+"-->当前偏移量："+offset);
    }

    /**
     * 消费者组mentuQueueGroup,方法listenerMentuQueue3
     * @param record
     */
    @KafkaListener(topics = {"mentuQueue"},groupId = "mentuQueueGroup")
    public void listenerMentuQueue3(ConsumerRecord<String,String> record){
        //获取消息
        String message = record.value();
        //消息偏移量
        long offset = record.offset();
        System.out.println("listenerMentuQueue3-->读取的消息："+message+"，分区："+record.partition()+"-->当前偏移量："+offset);
    }

    /**
     * 消费者组mentuQueueGroup,方法listenerMentuQueue4
     * @param record
     */
    @KafkaListener(topics = {"mentuQueue"},groupId = "mentuQueueGroup")
    public void listenerMentuQueue4(ConsumerRecord<String,String> record){
        //获取消息
        String message = record.value();
        //消息偏移量
        long offset = record.offset();
        System.out.println("listenerMentuQueue4-->读取的消息："+message+"，分区："+record.partition()+"-->当前偏移量："+offset);
    }
}
